package com.cl.ks.flow.handler;

import com.alibaba.fastjson.JSONObject;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.utils.DbUtil;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.jandar.ds.pojo.DbInfo;
import com.querydsl.jpa.impl.JPAQueryFactory;
import org.apache.spark.sql.Row;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;


@Component
public class SqlQueryHandler extends BaseFlowNodeHandler {
    private final DbUtil dbUtil;
    private final JPAQueryFactory queryFactory;


    public SqlQueryHandler(DbUtil dbUtil, JPAQueryFactory queryFactory) {
        this.dbUtil = dbUtil;
        this.queryFactory = queryFactory;
    }

    public String modifySqlLimit(String sql, Integer limit) {
        sql = sql.toLowerCase();
        // 使用正则表达式检查是否存在limit子句
        Pattern pattern = Pattern.compile("\\blimit\\b\\s+\\d+");
        Matcher matcher = pattern.matcher(sql.toLowerCase());

        if (matcher.find()) {
            return matcher.replaceFirst("limit " + limit);
        } else {
            if (sql.endsWith(";")) {
                return sql.substring(0, sql.length() - 1) + " limit " + limit + ";";
            } else {
                return sql + " limit " + limit + ";";
            }
        }
    }

    public List<JSONObject> previewSqlQuery(String dbCode, String sql) {
        if (dbCode == null || sql == null) {
            throw new RuntimeException("参数错误");
        }
        sql = modifySqlLimit(sql,100);
        DbInfo dbInfo = dbUtil.getDbInfo(dbCode);
        SparkParam sparkParam = new SparkParam();
        JSONObject expression = new JSONObject();
        expression.put("url", dbInfo.getUrl());
        expression.put("username", dbInfo.getUsername());
        expression.put("password", dbInfo.getPassword());
        expression.put("sql", sql);
        sparkParam.setNodeExpression(expression);
        SparkResult sparkResult = sparkNodeFactory.getSparkNodeByCode("/sql/query").handle(sparkParam);

        return sparkResult.getDataset().collectAsList().stream().map(Row::json).map(JSONObject::parseObject).collect(Collectors.toList());
    }

    @Override
    public SparkResult process(ProcessParam processParam) {

        JSONObject expression = processParam.getSparkNode().getNodeExpression();
        String sparkNodeName = "";
        String dbCode = String.valueOf(expression.get("dbCode"));
        String tableName = String.valueOf(expression.get("tableName"));
        String sql = String.valueOf(expression.get("sql"));
        String name = String.valueOf(expression.get("name"));
        sparkNodeName = "sql查询：" + dbCode + "-" + sql;

        if (dbCode == null) {
            throw new RuntimeException("未配置库表");
        }

        if (processParam.getTestFlag() == true || StringUtils.hasLength(processParam.getKsFlowGraph().getFailedNodeCode())) {
            sql = modifySqlLimit(sql,0);
        }

        DbInfo dbInfo = dbUtil.getDbInfo(dbCode);
        SparkParam sparkParam = new SparkParam();
        expression.put("url", dbInfo.getUrl());
        expression.put("username", dbInfo.getUsername());
        expression.put("password", dbInfo.getPassword());
        expression.put("sql", sql);
        sparkParam.setNodeExpression(expression);

        String power = "," +  processParam.getKsFlowGraph().getPower() +","+ processParam.getKsFlowGraph().getResultPower();
        Integer sourceConfigId = dbUtil.createSourceConfigAndMapping(dbCode, tableName, power,name+"(SQL查询)");
        processParam.getKsFlowGraph().getSourceConfigIdList().add(String.valueOf(sourceConfigId));


        SparkResult sparkResult = sparkNodeFactory.getSparkNodeByCode("/sql/query").handle(sparkParam);
        return sparkResult;
    }

    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.SQL_QUERY;
    }
}
